﻿using System;
using System.Collections.Generic;
using System.Configuration;
using System.Data;
using System.Data.SqlClient;
using System.Linq;
using System.ServiceModel;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Transactions;
using GrayParrot.Configuration;
using GrayParrot.Core.ServiceBus;
using GrayParrot.ServiceBroker.Configuration;

namespace GrayParrot.ServiceBroker
{
    public abstract class Subscriber<T> : IDisposable, ISubscriber<string> where T : class, new()
    {
        public event EventHandler<Exception> OnConnectionFailed;
        public event EventHandler OnStarted;
        public event EventHandler OnStopped;
        //public event EventHandler<Exception> OnError;

        public string Endpoint { get; set; }
        public string TopicSend { get; set; }
        public string TopicReceive { get; set; }
        public string TopicReceiveDeadLetter { get; set; }
        public string Subscription { get; set; }
        public bool ReadDL { get; set; }
        public UInt16 NumberOfRetriesOnFailure { get; set; }

        private bool _disposed;
        //private volatile bool m_shouldStop;
        // Backing fields for the ServiceStopRequested property.
        private static readonly object LockFlag = new object();
        private bool m_ServiceStopRequested = false;
        private string m_connectionString = string.Empty;
        private Subscription m_currentSubscription = null;
        private string m_currentTopicReceive = string.Empty;

        private bool ServiceStopRequested
        {
            get
            {
                lock (LockFlag)
                {
                    return m_ServiceStopRequested;
                }
            }
            set
            {
                lock (LockFlag)
                {
                    m_ServiceStopRequested = value;
                }
            }
        }

        public Subscriber()
        {
            //Init();
        }

        public Subscriber(string subscription)
        {
            ConfigurationManagerHelper pp = new ConfigurationManagerHelper();
            m_currentSubscription = pp.GetSubscription(subscription);

            Endpoint = m_currentSubscription.ConnectionString;
            TopicSend = m_currentSubscription.TopicSend;
            TopicReceive = m_currentSubscription.TopicReceive;
            TopicReceiveDeadLetter = m_currentSubscription.TopicReceiveDeadLetter;
            Subscription = subscription;
            ReadDL = m_currentSubscription.ReadDL;
            NumberOfRetriesOnFailure = m_currentSubscription.NumberOfRetriesOnFailure;

            m_connectionString = ConfigurationManager.ConnectionStrings[Endpoint].ConnectionString;

            //m_shouldStop = false;
            ServiceStopRequested = false;
        }

        public Subscriber(string endpoint, string subscription)
        {
            ConfigurationManagerHelper pp = new ConfigurationManagerHelper();
            m_currentSubscription = pp.GetSubscription(subscription);

            Endpoint = endpoint;
            TopicSend = m_currentSubscription.TopicSend;
            TopicReceive = m_currentSubscription.TopicReceive;
            TopicReceiveDeadLetter = m_currentSubscription.TopicReceiveDeadLetter;
            Subscription = subscription;
            NumberOfRetriesOnFailure = m_currentSubscription.NumberOfRetriesOnFailure;

            m_connectionString = ConfigurationManager.ConnectionStrings[Endpoint].ConnectionString;

            //m_shouldStop = false;
            ServiceStopRequested = false;
            //Init();
        }

        public void RequestStop()
        {
            //m_shouldStop = true;
            ServiceStopRequested = true;
        }

        public void Init()
        {
            try
            {
                if (!ReadDL)
                {
                    m_currentTopicReceive = TopicReceive;
                    GrayParrot.Logging.Logger.For<T>().Info("Create connection...");
                }
                else
                {
                    m_currentTopicReceive = TopicReceiveDeadLetter;
                    GrayParrot.Logging.Logger.For<T>().Info("Create connection to DL...");
                }

                if (Endpoint != null)
                {
                    StartListen();
                }
                else
                {
                    GrayParrot.Logging.Logger.For<T>().Error("Connection to Service Broker Failed !");

                    if (OnConnectionFailed != null)
                    {
                        OnConnectionFailed(this, new Exception("Connection to Service Broker Failed !"));
                    }
                }
            }
            catch (Exception ex)
            {
                GrayParrot.Logging.Logger.For<T>().Error("Connection to Service bus Failed !", ex);

                if (OnConnectionFailed != null)
                {
                    OnConnectionFailed(this, ex);
                }
            }
        }

        public void StartListen()
        {
            DateTime lastLogTime = DateTime.Now;
            string lastMessage = String.Empty;
            string msg = null;
            UInt16 m_errorCounter = 0;

            try
            {
                if (Endpoint != null)
                {
                    if (OnStarted != null)
                    {
                        OnStarted(this, EventArgs.Empty);
                    }

                    GrayParrot.Logging.Logger.For<T>().Info("... connection established");

                    // Continuously process messages received from the subscription 
                    while (!this.ServiceStopRequested)
                    {
                        using (TransactionScope scope = new TransactionScope())
                        {
                            using (SqlConnection conn = new SqlConnection(m_connectionString))
                            {
                                //using (SqlCommand cmd = new SqlCommand("[dbo].[ReceiveOdds]", conn))
                                using (SqlCommand cmd = new SqlCommand(m_currentTopicReceive, conn))
                                {
                                    cmd.CommandType = CommandType.StoredProcedure;
                                    cmd.CommandTimeout = 600; // seconds
                                    cmd.Parameters.Add("msg", SqlDbType.VarChar, -1).Direction = ParameterDirection.Output;

                                    try
                                    {
                                        conn.Open();
                                        cmd.ExecuteNonQuery();
                                        msg = cmd.Parameters["msg"].Value.ToString();

                                        Receive(msg);

                                        scope.Complete();

                                        m_errorCounter = 0;
                                    }
                                    catch (Exception ex)
                                    {
                                        GrayParrot.Logging.Logger.For<T>().Error(ex);

                                        m_errorCounter++;

                                        if (m_errorCounter == NumberOfRetriesOnFailure)
                                        {
                                            /* 
                                             * raggiunto il numero max di ripetizione del messaggio
                                             * lo spostiamo d'ufficio nella DL associata...
                                             */
                                            SendToDeadLetter(msg);

                                            /* procedo per non bloccare lo scodamento dei messaggi... */
                                            scope.Complete();

                                            /* reset del contatore... */
                                            m_errorCounter = 0;
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
            catch (TimeoutException timeoEx)
            {
                GrayParrot.Logging.Logger.For<T>().Error(timeoEx);

                throw;

                //EstablishConnection();
            }
            catch (UnauthorizedAccessException uaEx)
            {
                GrayParrot.Logging.Logger.For<T>().Error(uaEx);

                throw;
                //EstablishConnection();
            }
            catch (CommunicationException cmmEx)
            {
                GrayParrot.Logging.Logger.For<T>().Error(cmmEx);

                throw;
                //EstablishConnection();
            }
            catch (Exception ex)
            {
                GrayParrot.Logging.Logger.For<T>().Error(ex);

                throw;
                //EstablishConnection();
            }
        }

        private void SendToDeadLetter(string msg)
        {
            try
            {
                Publisher<T> m_pub = new Publisher<T>(m_currentSubscription);
                m_pub.SendToDeadLetter(msg);
                m_pub.Dispose();
                m_pub = null;
            }
            catch (Exception ex)
            {
                GrayParrot.Logging.Logger.For<T>().Error("Accodamento DL fallito", ex);
            }
        }

        private void EstablishConnection()
        {
            // wait 5 seconds and try to reconnect...
            Thread.Sleep(5000);

            Endpoint = null;

            // 
            Init();
        }

        public abstract bool Receive(string message);

        #region Dispose...
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        ~Subscriber()
        {
            Dispose(false);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (_disposed)
                return;

            if (disposing)
            {
                // free other managed objects that implement
                // IDisposable only
            }

            // release any unmanaged objects
            // set the object references to null
            Endpoint = string.Empty;
            TopicSend = string.Empty;
            TopicReceive = string.Empty;
            TopicReceiveDeadLetter = string.Empty;
            Subscription = string.Empty;
            m_currentSubscription = null;

            _disposed = true;
        }
        #endregion
    }
}
